package sql;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;

import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.sql.Timestamp;
import java.time.Duration;

import static org.apache.flink.table.api.Expressions.row;


/**
 *
 * 滚动窗口示例
 *
 1. The window function requires the timecol is a time attribute type, but is TIMESTAMP(6).
 *
 *
 */
public class D1Tumble {


    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        String genSql = "CREATE TABLE ods_tb ( " +
                " stime TIMESTAMP(3)," +
                " name STRING," +
                " id BIGINT," +
                " ts AS localtimestamp," +
                " rowtime AS localtimestamp," +
                " WATERMARK FOR rowtime AS rowtime - interval '30' second" +
                ") WITH ( " +
                " 'connector'='datagen', " +
                " 'rows-per-second'='1'," +
                " 'fields.name.length'='1'," +
                " 'fields.id.kind'='sequence'," +
                " 'fields.id.start'='1'," +
                " 'fields.id.end'='10000000'" +
                ")";


        /**
         * 使用 tumble 会自动生成：
         * window_start、window_end、window_time
         */
        String viewSql = "CREATE VIEW v_base AS " +
                " SELECT " +
                " name,count(1) as cnt" +
                " ,window_start,window_end,window_time " +
                " from " +
                " table(tumble(table ods_tb, descriptor(rowtime), interval '1' minutes))" +
                " GROUP BY name,window_start,window_end,window_time";

        System.out.println(viewSql);


        String sink = "CREATE TABLE pprint (" +"    " +
                "    ntime STRING, " +
                "    name STRING, " +
                "    cnt BIGINT " +
                "    ,window_start TIMESTAMP(3) " +
                "    ,window_end TIMESTAMP(3) " +
                "    ,window_time TIMESTAMP(3) " +
                ") WITH (" +
                "    'connector' = 'print'" +
                ")";
        String sinkSql = "INSERT INTO pprint " +
                "SELECT DATE_FORMAT(CURRENT_TIMESTAMP, 'yyyy-MM-dd HH:mm:ss')," +
                " name,cnt" +
                " ,window_start,window_end,window_time" +
                " FROM v_base";




        tableEnv.executeSql(genSql);
        tableEnv.executeSql(viewSql);
        tableEnv.executeSql("DESC v_base").print();
        tableEnv.executeSql(sink);
        tableEnv.executeSql(sinkSql);
        System.out.println(new Timestamp(System.currentTimeMillis()));



    }
}
